package com.heima.kafka.stream;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {
    @Bean
    public KStream<String,String> kStream(StreamsBuilder builder){
        //获取数据，指定topic名字
        KStream<String, String> stream = builder.stream("streams-test-107");
        //计算方法写法和stream流处理基本一致
        //value "hello kafka"  -> [value : "hello",value : "kakfa"]
        stream.flatMapValues(value -> Arrays.asList(value.split(" ")))
                //按value分组
                .groupBy((key, value) -> value)
                //设置统计时间窗口->每隔多少时间统计一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                //key : "hello"  value："hello" -> count
                .count()
                //key : hello value: 2 将KTABLE转换成KSTREAM
                .toStream()
                //将key value调整成String类型
                .map((key, value) -> new KeyValue<>(key.key(), value.toString()))
                //投递结果到kafka
                .to("streams-out-107");
        return stream;
    }
}